package com.newbee.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 测试mapreduce连接hbase
 *
 * @author yujie
 * @create 2017-12-14 上午9:29
 **/
public class HBaseJob {

    private static AtomicInteger integer = new AtomicInteger(1);

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = HBaseConfiguration.create();
        Job job = Job.getInstance(conf, "connect_hbase");
        //TableMapReduceUtil.addDependencyJars(job);

        /**
         * hbase 强依赖 zookeeper 客户端直接连接zk
         */
//        conf.set("hbase.cluster.distributed", "true");
//        conf.set("zookeeper.znode.parent", "/hbase");
//        conf.set("hbase.zookeeper.quorum", "hbase.240.155.hadoop.sjz,hbase.240.156.hadoop.sjz,hbase.243.6.hadoop.sjz");

        conf.set("hbase.zookeeper.quorum","localhost");
        /**
         * 创建scan
         */
        Scan scan = new Scan();
        Filter row1 = new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new BinaryComparator("r1".getBytes()));
        scan.addFamily("id".getBytes());
        scan.addFamily("name".getBytes());
        scan.setFilter(row1);

        TableMapReduceUtil.initTableMapperJob("NB:users".getBytes(), scan, MyMapper.class, Text.class, Text.class, job);
        TableMapReduceUtil.initTableReducerJob("NB:copy_users", MyTableReducer.class, job);
//        job.setOutputFormatClass(NullOutputFormat.class);
//        job.setReducerClass(MyIncrReducer.class);
//        job.setOutputKeyClass(NullWritable.class);
//        job.setOutputValueClass(NullWritable.class);
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

    public static class MyMapper extends TableMapper<Text, Text> {


        public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
            String id = new String(value.getValue("id".getBytes(), "id".getBytes()));
            String name = new String(value.getValue("name".getBytes(), "first".getBytes()));
            context.write(new Text(id), new Text(name));
        }
    }


    public static class MyTableReducer extends TableReducer<Text, Text, NullWritable> {

        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            Put put = new Put(("r" + integer.get()).getBytes());
            put.addColumn("id".getBytes(), "id".getBytes(), key.toString().getBytes());
            put.addColumn("name".getBytes(), "first".getBytes(), values.iterator().next().toString().getBytes());
            integer.getAndIncrement();
            context.write(NullWritable.get(), put);
        }
    }

    public static class MyIncrReducer extends Reducer<Text, Text, NullWritable, NullWritable> {

        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            Connection connection = ConnectionFactory.createConnection(context.getConfiguration());
            Table table = connection.getTable(TableName.valueOf("counters".getBytes()));
            table.incrementColumnValue("r1".getBytes(), "name".getBytes(), "name".getBytes(), 1);
            table.close();
            connection.close();
            context.write(NullWritable.get(), NullWritable.get());
        }
    }

}
